//package cn.com.zetatech.loader.resolver.kafka.consumer;
//
//import cn.com.zetatech.loader.entity.ParquetMessage;
//import cn.com.zetatech.loader.entity.ResolverBinMapMessage;
//import cn.com.zetatech.loader.resolver.service.ResolverParquetFunction;
//import com.google.gson.Gson;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.core.KafkaTemplate;
//import org.springframework.kafka.support.Acknowledgment;
//import org.springframework.stereotype.Component;
//
//import java.util.List;
//
///**
// * @author jian.chen@zetatech.com.cn
// * @Description TODO
// * @Date 2024/8/18 14:59
// */
//@Slf4j
//@Component
//public class ParquetResolveConsumer {
//
//    @Autowired
//    private Gson gson;
//
//    @Autowired
//    private List<ResolverParquetFunction> functions;
//
//    @KafkaListener(topics = "#{'${resolver.topic}'.split(',')}",
//            concurrency = "${resolver.consumer.concurrency:2}",
//            groupId = "${resolver.consumer.groupId:parquet_consume3 }")
//    public void onMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {
//        try {
//            String value = record.value();
//            String topic = record.topic();
//            String key = record.key();
//
//            System.out.println(topic);
//            System.out.println(key);
//            if(!StringUtils.isEmpty(value)) {
//                // 对消息进行解码
//                log.info("<============================ 消费Kafka parquet消息  ============================>");
//                ParquetMessage message = gson.fromJson(value, ParquetMessage.class);
//                if(message != null) {
//                    log.info("<============================ 开始解析任务 ============================>");
//                    for (ResolverParquetFunction function : functions) {
//                        function.process(message);
//                    }
//                }
//            }
//        } catch (Exception e) {
//            log.error(e.getMessage(), e);
//        } finally {
//            ack.acknowledge();
//        }
//    }
//}
